package com.runssnail.demo.kafka.jms.spring;

import com.runssnail.demo.kafka.jms.KafkaTopicSubscriber;
import kafka.consumer.ConsumerConfig;
import kafka.javaapi.consumer.ConsumerConnector;
import org.apache.commons.lang3.Validate;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;

import javax.jms.MessageListener;
import javax.jms.TopicSubscriber;
import java.util.List;
import java.util.Properties;

/**
 * TopicSubscriber FactoryBean
 *
 * <p/>
 * Created by zhengwei on 2015/8/28.
 */
public class TopicSubscriberFactoryBean implements FactoryBean<TopicSubscriber>, InitializingBean, DisposableBean {

    private TopicSubscriber topicSubscriber;

    /**
     * 订阅主题，多个用','分割
     */
    private String topics;

    /**
     * 配置
     */
    private Properties configs;

    /**
     * 消息监听器
     */
    private List<MessageListener> messageListeners;

    @Override
    public TopicSubscriber getObject() throws Exception {
        return this.topicSubscriber;
    }

    @Override
    public Class<?> getObjectType() {
        return TopicSubscriber.class;
    }

    @Override
    public boolean isSingleton() {
        return true;
    }

    @Override
    public void afterPropertiesSet() throws Exception {

        Validate.notNull(this.topics, "topics is required");
        Validate.notEmpty(this.configs, "configs is required");
        Validate.notEmpty(this.messageListeners, "messageListeners is required");

        ConsumerConfig config = new ConsumerConfig(this.configs);

        ConsumerConnector consumer = kafka.consumer.Consumer.createJavaConsumerConnector(config);

        KafkaTopicSubscriber kafkaTopicSubscriber = new KafkaTopicSubscriber(topics, consumer);
        kafkaTopicSubscriber.setMessageListeners(this.messageListeners);
        this.topicSubscriber = kafkaTopicSubscriber;

        kafkaTopicSubscriber.receive();

    }

    /**
     * 主题，多个主题用逗号分隔
     *
     * @param topics 主题
     */
    public void setTopics(String topics) {
        this.topics = topics;
    }

    /**
     * kafka 配置
     *
     * @param configs kafka 配置
     */
    public void setConfigs(Properties configs) {
        this.configs = configs;
    }

    /**
     * 消息监听器，有消息来时，会通知监听器
     *
     * @param messageListeners 消息监听器
     */
    public void setMessageListeners(List<MessageListener> messageListeners) {
        this.messageListeners = messageListeners;
    }

    @Override
    public void destroy() throws Exception {
        this.topicSubscriber.close();
    }
}
